package com.google.appengine.examples;

import com.google.appengine.api.files.Crc32c;
import com.google.appengine.api.files.FileReadChannel;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;

/**
 * This is Java a version of RecordReader in Python
 * {google/appengine/api/files/records.RecordReader} which allows reading a file written in leveldb
 * format: http://leveldb.googlecode.com/svn/trunk/doc/log_format.txt This class is nothing more
 * than a public copy of {@link com.google.appengine.api.files.RecordReadChannelImpl} and changes in
 * file API might leave it as redundant.
 */
public class RecordReader {
  public static final int BLOCK_SIZE = 32 * 1024;
  public static final int HEADER_LENGTH = 7;
  public static final int CRC_MASK_DELTA = 0xa282ead8;
  private final FileReadChannel inputChannel;
  private final ByteBuffer blockBuffer;
  private ByteBuffer finalRecord;

  private final Object lock = new Object();

  /**
   * An enumerated type that describes the type of the record being stored. These bytes much leveldb
   * log format.
   */
  public enum RecordType {
    NONE((byte) 0x00),
    FULL((byte) 0x01),
    FIRST((byte) 0x02),
    MIDDLE((byte) 0x03),
    LAST((byte) 0x04),
    UNKNOWN((byte) 0xFF);

    private final byte value;

    private RecordType(byte value) {
      this.value = value;
    }

    /**
     * The byte value of the record type is written to the file as part of the header.
     *
     * @return the byte value of the record type.
     */
    public byte value() {
      return this.value;
    }

    /**
     * Converts a byte value into a {@link RecordType} enum.
     *
     * @param value the byte value of the {@link RecordType} you want.
     * @return a {@link RecordType} that corresponds to the inputed byte value.
     */
    public static RecordType get(byte value) {
      switch (value) {
        case 0x00:
          return NONE;
        case 0x01:
          return FULL;
        case 0x02:
          return FIRST;
        case 0x03:
          return MIDDLE;
        case 0x04:
          return LAST;
        default:
          return UNKNOWN;
      }
    }
  }

  private static final class Record {
    private final ByteBuffer data;
    private final RecordType type;

    public Record(RecordType type, ByteBuffer data) {
      this.type = type;
      this.data = data;
    }

    public ByteBuffer data() {
      return this.data;
    }

    public RecordType type() {
      return this.type;
    }
  }

  class RecordReadException extends Exception {
    public RecordReadException(String errorMessage) {
      super(errorMessage);
    }
  }

  public static long maskCrc(long crc) {
    return ((((crc >> 15) | (crc << 17)) + CRC_MASK_DELTA) & 0xFFFFFFFFL);
  }

  public static long unmaskCrc(long maskedCrc) {
    long rot = (maskedCrc - CRC_MASK_DELTA) & 0xFFFFFFFFL;
    return (((rot >> 17) | (rot << 15)) & 0xFFFFFFFFL);
  }

  public RecordReader(FileReadChannel inputChannel) {
    this.inputChannel = inputChannel;
    blockBuffer = ByteBuffer.allocate(BLOCK_SIZE);
    blockBuffer.order(ByteOrder.LITTLE_ENDIAN);
    finalRecord = ByteBuffer.allocate(BLOCK_SIZE);
    finalRecord.order(ByteOrder.LITTLE_ENDIAN);
  }

  public ByteBuffer readRecord() throws IOException {
    synchronized (lock) {
      finalRecord.clear();
      RecordType lastRead = RecordType.NONE;
      while (true) {
        try {
          Record record = readPhysicalRecord();
          if (record == null) {
            return null;
          }
          switch (record.type()) {
            case NONE:
              validateRemainderIsEmpty();
              break;
            case FULL:
              if (lastRead != RecordType.NONE) {
                throw new RecordReadException("Invalid RecordType: " + record.type);
              }
              return record.data().slice();
            case FIRST:
              if (lastRead != RecordType.NONE) {
                throw new RecordReadException("Invalid RecordType: " + record.type);
              }
              finalRecord = appendToBuffer(finalRecord, record.data());
              break;
            case MIDDLE:
              if (lastRead == RecordType.NONE) {
                throw new RecordReadException("Invalid RecordType: " + record.type);
              }
              finalRecord = appendToBuffer(finalRecord, record.data());
              break;
            case LAST:
              if (lastRead == RecordType.NONE) {
                throw new RecordReadException("Invalid RecordType: " + record.type);
              }
              finalRecord = appendToBuffer(finalRecord, record.data());
              finalRecord.flip();
              return finalRecord.slice();
            default:
              throw new RecordReadException("Invalid RecordType: " + record.type.value());
          }
          lastRead = record.type();
        } catch (RecordReadException e) {
          finalRecord.clear();
          sync();
        }
      }
    }
  }

  /**
   * Reads the next record from the RecordIO data stream.
   *
   * @return Record data about the physical record read.
   * @throws IOException
   */
  private Record readPhysicalRecord() throws IOException, RecordReadException {
    int bytesToBlockEnd = (int) (BLOCK_SIZE - (inputChannel.position() % BLOCK_SIZE));

    if (bytesToBlockEnd < HEADER_LENGTH) {
      return new Record(RecordType.NONE, null);
    }

    blockBuffer.clear();
    blockBuffer.limit(HEADER_LENGTH);
    int bytesRead = inputChannel.read(blockBuffer);
    if (bytesRead != HEADER_LENGTH) {
      return null;
    }
    blockBuffer.flip();
    int checksum = blockBuffer.getInt();
    short length = blockBuffer.getShort();
    RecordType type = RecordType.get(blockBuffer.get());
    if (length > bytesToBlockEnd || length < 0) {
      throw new RecordReadException("Length is too large:" + length);
    }

    blockBuffer.clear();
    blockBuffer.limit(length);
    bytesRead = inputChannel.read(blockBuffer);
    if (bytesRead != length) {
      return null;
    }
    if (!isValidCrc(checksum, blockBuffer, type.value())) {
      throw new RecordReadException("Checksum doesn't validate.");
    }

    blockBuffer.flip();
    return new Record(type, blockBuffer);
  }

  /**
   * Validates that the {@link Crc32c} validates.
   *
   * @param checksum the checksum in the record.
   * @param data the {@link ByteBuffer} of the data in the record.
   * @param type the byte representing the {@link RecordType} of the record.
   * @return true if the {@link Crc32c} validates.
   */
  private boolean isValidCrc(int checksum, ByteBuffer data, byte type) {
    if (checksum == 0 && type == 0 && data.limit() == 0) {
      return true;
    }
    Crc32c crc = new Crc32c();
    crc.update(type);
    crc.update(data.array(), 0, data.limit());

    return unmaskCrc(checksum) == crc.getValue();
  }

  private void validateRemainderIsEmpty() throws IOException, RecordReadException {
    int bytesToBlockEnd = (int) (BLOCK_SIZE - (inputChannel.position() % BLOCK_SIZE));
    blockBuffer.clear();
    blockBuffer.limit(bytesToBlockEnd);
    int read = inputChannel.read(blockBuffer);
    if (read != bytesToBlockEnd) {
      throw new RecordReadException(
          "There are " + bytesToBlockEnd + " but " + read + " were read.");
    }
    blockBuffer.flip();
    for (int i = 0; i < bytesToBlockEnd; i++) {
      byte b = blockBuffer.get(i);
      if (b != 0) {
        throw new RecordReadException("Found a non-zero byte: " + b
            + " before the end of the block " + i
            + " bytes after encountering a RecordType of NONE");
      }
    }
  }

  /**
   * Appends a {@link ByteBuffer} to another. This may modify the inputed buffer that will be
   * appended to.
   *
   * @param to the {@link ByteBuffer} to append to.
   * @param from the {@link ByteBuffer} to append.
   * @return the resulting appended {@link ByteBuffer}
   */
  private static ByteBuffer appendToBuffer(ByteBuffer to, ByteBuffer from) {
    if (to.remaining() < from.remaining()) {
      int capacity = to.capacity();
      while (capacity - to.position() < from.remaining()) {
        capacity *= 2;
      }
      ByteBuffer newBuffer = ByteBuffer.allocate(capacity);
      to.flip();
      newBuffer.put(to);
      to = newBuffer;
      to.order(ByteOrder.LITTLE_ENDIAN);
    }
    to.put(from);
    return to;
  }

  /**
   * Moves to the start of the next block.
   * @throws IOException
   */
  private void sync() throws IOException {
    long padLength = BLOCK_SIZE - (inputChannel.position() % BLOCK_SIZE);
    inputChannel.position(inputChannel.position() + padLength);
  }
}
